package com.qf.listener;

import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Component;

import java.util.concurrent.TimeUnit;

/**
 * 防止重复消费业务实现
 * @author 千锋健哥
 */
@Component
@RabbitListener(queues = "queue1")
public class Queue1Consumer {

    @Autowired
    private StringRedisTemplate redisTemplate;

    @RabbitHandler
    public void messageHandler(String msg, Channel channel, Message message) throws Exception {

        //1. 获取消息唯一编号
        long msgId = message.getMessageProperties().getDeliveryTag();

        //2. 使用setnx命令将数据存入redis中, key是消息的唯一编号, value值是1, 设置超时时间10秒
        Boolean flag = redisTemplate.opsForValue().setIfAbsent(String.valueOf(msgId), "1", 10, TimeUnit.SECONDS);

        if (flag) {
            //3. 如果能够存入redis, 那么处理业务
            System.out.println("======正常处理业务======");
            //手动ack, 返回true, 处理成功
            /**
             * 第一个参数: DeliveryTag,消息的唯一标识  channel+消息编号
             * 第二个参数：是否开启批量处理 false:不开启批量
             * 举个栗子： 假设我先发送三条消息deliveryTag分别是5、6、7，可它们都没有被确认，
             *           我发第四条消息此时deliveryTag为8，multiple设置为 true，
             *           会将5、6、7、8的消息全部进行确认。
             */
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        } else {
            //4. 如果无法存入redis, 那么说明之前处理过这个数据的业务, 直接返回ack给mq
            /**
             * 第一个参数: DeliveryTag,消息的唯一标识  channel+消息编号
             * 第二个参数：是否开启批量处理 false:不开启批量
             * 举个栗子： 假设我先发送三条消息deliveryTag分别是5、6、7，可它们都没有被确认，
             *           我发第四条消息此时deliveryTag为8，multiple设置为 true，
             *           会将5、6、7、8的消息全部进行确认。
             */
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        }


    }
}
